//本代码由汪碧康编写于2014年12月17日
//pconline.com.cn
//用于通过消息队列httpsqs或rabbitmq或者http请求的方式来清除apache traffic server中的缓存
// mqac.go mesage queue clean ats cache
//v2.0

package main

import (
	"flag"
	"fmt"
	"github.com/streadway/amqp"
	"goconfig"
	"io"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	//	"strconv"
	"strings"
	"time"
)

var cfpath = flag.String("f", "", "Specifies the configuration file path")

type Conf struct {
	AtsIp           string
	MqType          string
	RbmqIp          string
	RbmqAdmin       string
	RbmqPass        string
	RbmqVhost       string
	RbmqQueue       string
	MaxMessage      float64
	HttpsqsIp       string
	HttpsqsQueue    string
	HttpPort        string
	ExchangeName    string
	RouteKey        string
	ReadInterval    int64
	CdnCleanCache   bool
	CdnInterfaceUrl string
	ProxyAdress     string
}

//getconfig
func GetConfig() (c *Conf) {
	flag.Parse()
	gr, err := goconfig.ReadConfigFile(*cfpath)
	if err != nil {
		fmt.Println("ERR:load etc/mqcac.conf config filure!")
		os.Exit(1)
	}
	c = &Conf{}
	var Section string = "GLOBAL"
	var err2 error
	c.AtsIp, _ = gr.GetString(Section, "AtsIp")
	c.MqType, _ = gr.GetString(Section, "MqType")
	c.RbmqIp, _ = gr.GetString(Section, "RbmqIp")
	c.RbmqAdmin, _ = gr.GetString(Section, "RbmqAdmin")
	c.RbmqPass, _ = gr.GetString(Section, "RbmqPass")
	c.RbmqVhost, _ = gr.GetString(Section, "RbmqVhost")
	c.RbmqQueue, _ = gr.GetString(Section, "RbmqQueue")
	c.ExchangeName, _ = gr.GetString(Section, "ExchangeName")
	c.RouteKey, _ = gr.GetString(Section, "RouteKey")

	c.CdnCleanCache, _ = gr.GetBool(Section, "CdnCleanCache")
	c.CdnInterfaceUrl, _ = gr.GetString(Section, "CdnInterfaceUrl")
	c.ProxyAdress, _ = gr.GetString(Section, "ProxyAdress")

	c.MaxMessage, _ = gr.GetFloat(Section, "MaxMessage")
	if c.MaxMessage <= 0 {
		c.MaxMessage = 2
	}
	c.ReadInterval, _ = gr.GetInt64(Section, "ReadInterval")
	if c.ReadInterval <= 0 {
		c.ReadInterval = 500
	}

	c.HttpsqsIp, _ = gr.GetString(Section, "HttpsqsIp")
	c.HttpsqsQueue, _ = gr.GetString(Section, "HttpsqsQueue")

	c.HttpPort, err2 = gr.GetString(Section, "HttpPort")
	if err2 != nil {
		log.Printf("ERR:GetConfig failure,err%s:", err2)
	} else {
		if len(c.HttpPort) <= 0 {
			c.HttpPort = "8800"
		}
	}
	return c
}

//proxy is atsip
func GetByProxy(url_addr, proxy_addr, method string) (*http.Response, error) {
	proxy_addr = "http://" + proxy_addr
	request, _ := http.NewRequest(method, url_addr, nil)
	proxy, err := url.Parse(proxy_addr)
	if err != nil {
		return nil, err
		log.Printf("ERR:GetByProxy err,ip:%s,err:\n", proxy_addr, err)
	}
	client := &http.Client{
		Transport: &http.Transport{
			Proxy: http.ProxyURL(proxy),
			Dial: func(netw, proxy string) (net.Conn, error) {
				deadline := time.Now().Add(5 * time.Second)
				c, err := net.DialTimeout(netw, proxy, time.Second*5)
				if err != nil {
					return nil, err
				}
				c.SetDeadline(deadline)
				return c, nil
			},
		},
	}
	return client.Do(request)
}

func ClearAts(tsip string, xurl string) {
	xip := strings.Split(strings.TrimSpace(tsip), ",")
	if len(xip) > 0 {
		for _, ip := range xip {
			resp, err := GetByProxy(xurl, ip, "PURGE")
			if err != nil {
				log.Printf("ERR:GetByProxy failure!Error:", err)
				log.Printf("ERR:failure ip :%s\n", ip)
				return
			} else {
				defer resp.Body.Close()
				switch {
				case resp.StatusCode == 200:
					log.Printf("INFO:clear success %s,atsip=%s\n", xurl, ip)
					return
				case resp.StatusCode == 404:
					log.Printf("INFO:Not Found %s,atsip=%s\n", xurl, ip)
					return
				default:
					log.Printf("WARN:clear faiure,StatusCode=%d,atsip=%s,uri=%s", resp.StatusCode, ip, xurl)
				}
			}
		}
	}

}

//no cdn get url
func GetUrl(url, method string) (*http.Response, error) {
	request, _ := http.NewRequest(method, url, nil)
	client := &http.Client{
		Transport: &http.Transport{
			Dial: func(netw, proxy string) (net.Conn, error) {
				deadline := time.Now().Add(10 * time.Second)
				c, err := net.DialTimeout(netw, proxy, time.Second*5)
				if err != nil {
					return nil, err
				}
				c.SetDeadline(deadline)
				return c, nil
			},
		},
	}
	return client.Do(request)
}

//clean CDN cache
func ClearCdn(isclscdn bool, proxyip, cdnintf, xurl []string) {
	if isclscdn {
		var allurl string
		for _, url := range xurl {
			allurl = allurl + url + ";"
		}
		turl := cdnintf + strings.TrimRight(allurl, ";")
		isproxy := (proxyip != "")
		var resp *http.Response
		var err error
		if isproxy {
			resp, err = GetByProxy(turl, proxyip, "GET")
		} else {
			resp, err = GetUrl(turl, "GET")
		}
		if err != nil {
			log.Printf("ERR:GetUrl CDN failure!Error:", err)
			log.Printf("ERR:failure url :%s\n", turl)
			return
		} else {
			defer resp.Body.Close()
			switch {
			case resp.StatusCode == 200:
				log.Printf("INFO:CDN clear success %s\n", xurl)
				return
			case resp.StatusCode == 404:
				log.Printf("INFO:CDN Not Found %s\n", xurl)
				return
			default:
				log.Printf("WARN:CDN clear faiure,StatusCode=%d,uri=%s", resp.StatusCode, xurl)
			}
		}
	}

}

//get message form rabbitmq
func GetMessage(amqpURI, queueName, Routekey, Exchangename string, maxmsg int) (m []string, isok bool) {
	conn, err := amqp.Dial(amqpURI)
	if err != nil {
		log.Printf("ERR:Dial amqpURI failure!")
		return
	}
	defer func() {
		conn.Close()
	}()

	channel, err := conn.Channel()
	if err != nil {
		log.Printf("ERR:Conn Channel failure!")
	}

	if err = channel.QueueBind(queueName, Routekey, Exchangename, false, nil); err != nil {
		//fmt.Errorf("Queue Bind: %s", err)
		log.Printf("ERR:Queue Bind failure,err: %s", err)
	}
	//var isok bool = false
	for i := 0; i < maxmsg; i++ {
		msg, ok, err := channel.Get(queueName, false)
		if err != nil {
			log.Printf("ERR:Queue Get failure!:%s", err)
		}
		if ok {
			msg.Reject(false)
			t1 := strings.Split(string(msg.Body), ",")
			m = append(m, t1...)
			if len(m) > 0 {
				isok = true
			} else {
				isok = false
			}
		} else {
			break
		}
	}
	return
}

//get msg from httpsqs
func GetHttpsqsMsg(HttpsqsIp, PoolName string) (m []string, isok bool) {
	Url := "http://" + HttpsqsIp + "/?name=" + PoolName + "&opt=get"
	client := new(http.Client)
	reg, err := http.NewRequest("GET", Url, nil)
	if err != nil {
		log.Printf("ERR:Read httpsqs failure! err:%s\n", err)
		return
	}
	reg.Header.Set("Connection", "keep-alive")
	resp, err := client.Do(reg)
	if err != nil {
		log.Printf("ERR:Client.do failure! err:%s\n", err)
		return
	}
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Printf("ERR:get body failure! err:%s\n", err)
	}
	if string(body) != "HTTPSQS_GET_END" {
		m = strings.Split(string(body), ",")
		if len(m) > 0 {
			isok = true
		} else {
			isok = false
		}
	}
	return
}

func Handler(w http.ResponseWriter, r *http.Request) {
	r.ParseForm()
	if r.Method == "GET" {
		v := GetConfig()
		durl := r.Form.Get("url")
		xip := strings.Split(v.AtsIp, ";")
		for _, ip := range xip {
			go ClearAts(ip, durl)
		}
		io.WriteString(w, "success\n")
	}
}

func daemon() {
	v := GetConfig()
	RbmqUri := "amqp://" + v.RbmqAdmin + ":" + v.RbmqPass + "@" + v.RbmqIp + "/" + v.RbmqVhost
	xip := strings.Split(v.AtsIp, ";")
	MaxMessage := int(v.MaxMessage)
	ReadInterval := time.Duration(v.ReadInterval)
	httpport := ":" + v.HttpPort

	//var xurl = []string{"http://bbs.pcauto.com.cn/topic-6044255.html"}
	var xurl []string
	var yes bool
	go func() {
		for {
			if v.MqType == "rabbitmq" {
				xurl, yes = GetMessage(RbmqUri, v.RbmqQueue, v.RouteKey, v.ExchangeName, MaxMessage)
			} else if v.MqType == "httpsqs" {
				xurl, yes = GetHttpsqsMsg(v.HttpsqsIp, v.HttpsqsQueue)
			} else {
				log.Printf("ERR:unsupport the mqtype:%s,can't start\n", v.MqType)
				os.Exit(-1)
			}
			if yes {
				for _, ip := range xip {
					for _, url := range xurl {
						go ClearAts(ip, url)
						go ClearCdn(v.CdnCleanCache, v.ProxyAdress, v.CdnInterfaceUrl, url)
					}
				}
			}
			time.Sleep(ReadInterval * time.Millisecond)
		}
	}()

	http.HandleFunc("/", Handler)
	err1 := http.ListenAndServe(httpport, nil)
	if err1 != nil {
		log.Fatal("ERR:ListenAndServe: ", err1)
	}
}

func main() {
	daemon()
}
